/**
* Copyright © 2018 深圳市巨鼎医疗设备有限公司
*/
package com.zl.websocket.service.impl;

import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import com.zl.websocket.constants.WebsocketConstants;
import com.zl.websocket.entity.WebsocketMessage;
import com.zl.websocket.entity.WebsocketMessageTransform;
import com.zl.websocket.service.RouteService;

/**
 * websocket用户、应用实例路由关系处理服务
 * 
 * @author Zhouych
 * @Date: 2018年6月28日 下午4:32:38
 * @since JDK 1.8
 */
@Service("websocketRouteService")
@Scope(ConfigurableBeanFactory.SCOPE_SINGLETON)
public class WebsocketRouteServiceImpl implements RouteService {

	@Autowired
	private StringRedisTemplate stringRedisTemplate;

	@Autowired
	@Qualifier(value = "redisTemplate")
	private RedisTemplate<String, Object> redisTemplate;

	@Override
	public String getInstanceId() {
		return WebsocketConstants.WEBSOCKET_UUID;
	}

	@Override
	public String getInstanceIdByUserId(String userId) {
		if (StringUtils.isBlank(userId)) {
			return null;
		}
		return (String) stringRedisTemplate.opsForHash().get(WebsocketConstants.WEBSOCKET_USER_CHANNEL, userId);
	}

	@Override
	@Async
	public void userConnect(String userId) {
		if (StringUtils.isBlank(userId)) {
			return;
		}
		// 记录用户ID和channel对应关系，即标志该用户是连接到那个实例的
		stringRedisTemplate.opsForHash().put(WebsocketConstants.WEBSOCKET_USER_CHANNEL, userId, getInstanceId());
		// 把用户ID添加到该机器实例的用户列表
		stringRedisTemplate.opsForSet().add(getInstanceId(), userId);
	}

	@Override
	@Async
	public void userDisConnect(String userId) {
		if (StringUtils.isBlank(userId)) {
			return;
		}
		stringRedisTemplate.opsForHash().delete(WebsocketConstants.WEBSOCKET_USER_CHANNEL, userId);
		stringRedisTemplate.opsForSet().remove(getInstanceId(), userId);
	}

	@Override
	public void instanceInit() {
		checkExistAndClear();
		stringRedisTemplate.opsForSet().add(WebsocketConstants.WEBSOCKET_CHANNELS, getInstanceId());
	}

	@Override
	@Async
	public void clearAll(final String instanceId) {
		stringRedisTemplate.opsForSet().remove(WebsocketConstants.WEBSOCKET_CHANNELS, instanceId);
		Set<String> members = stringRedisTemplate.opsForSet().members(instanceId);
		if (null != members && !members.isEmpty()) {
			for (String member : members) {
				stringRedisTemplate.opsForHash().delete(WebsocketConstants.WEBSOCKET_USER_CHANNEL, member);
			}
		}
		stringRedisTemplate.delete(instanceId);
	}


	private void checkExistAndClear() {
		Set<String> channels = stringRedisTemplate.opsForSet().members(WebsocketConstants.WEBSOCKET_CHANNELS);
		if (null != channels && !channels.isEmpty()) {
			for (String channel : channels) {
				if (!checkExist(channel)) {
					clearAll(channel);
				}
			}
		}
	}

	@Override
	public boolean checkExist(String channel) {
		WebsocketMessageTransform testMsg = new WebsocketMessageTransform();
		WebsocketMessage payload = new WebsocketMessage();
		payload.setMessageType(-1);
		testMsg.setPayload(payload);
		final byte[] rawChannel = redisTemplate.getStringSerializer().serialize(channel);
		final byte[] rawMessage = ((GenericJackson2JsonRedisSerializer) redisTemplate.getValueSerializer())
				.serialize(testMsg);
		boolean exist = redisTemplate.execute(new RedisCallback<Boolean>() {
			public Boolean doInRedis(RedisConnection connection) {
				Long count = connection.publish(rawChannel, rawMessage);
				return !(null == count || count <= 0);
			}
		}, true);
		return exist;
	}
}
